package writer.file;


import org.apache.commons.io.FileUtils;
import util.InfoUtil;

import java.io.*;
import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

/**
 * @author kaddddd
 * @Time：2017年8月16日 下午5:26:37
 * @version 1.0
 * @description
 */
public class FileCreateTest {

    final static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) throws Exception {


        //        File f = new File("D://file");
//        // 文件总数
//        final List<File> filePathsList = new ArrayList<File>();
//        File[] filePaths = f.listFiles();
//        for (File s : filePaths) {
//            filePathsList.add(s);
//        }
//
//        CountDownLatch latch = new CountDownLatch(filePathsList.size());
//        ExecutorService pool = Executors.newFixedThreadPool(10);
//
//        BlockingQueue<Future<Map<String, FileInputStream>>> queue = new ArrayBlockingQueue<Future<Map<String, FileInputStream>>>(100);
//
//        System.out.println("-------------文件读、写任务开始时间：" + sdf.format(new Date()));
//        for (int i = 0; i < filePathsList.size(); i++) {
//            File temp = filePathsList.get(i);
//            Future<Map<String, FileInputStream>> future = pool.submit(new MyCallableProducer(latch, temp));
//            queue.add(future);
//
//            pool.execute(new MyCallableConsumer(queue));
//        }
//
//        try {
//            latch.await();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println("-------------文件读、写任务结束时间：" + sdf.format(new Date()));
//        pool.shutdownNow();

//        File f = new File("D://file");
//        // 文件总数
//        final List<File> filePathsList = new ArrayList<File>();
//        File[] filePaths = f.listFiles();
//        BlockingQueue<Map<String, FileInputStream>> queue = new ArrayBlockingQueue<Map<String, FileInputStream>>(100);
//
//        for (File s : filePaths) {
//            filePathsList.add(s);
//
//            FileInputStream fis = new FileInputStream(s);
//            Map<String, FileInputStream> fileMap = new HashMap<String, FileInputStream>();
//            fileMap.put(s.getName(), fis);
//            queue.add(fileMap);
//        }
//
//        CountDownLatch latch = new CountDownLatch(filePathsList.size());
//        ExecutorService pool = Executors.newFixedThreadPool(4);
//
//        System.out.println("-------------文件读开始时间：" + sdf.format(new Date()));
//        for (int i = 0; i < filePathsList.size(); i++) {
//            pool.execute(new MyCallableCon(queue, latch));
//        }
//
//        try {
//            latch.await();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println("-------------文件读、写任务结束时间：" + sdf.format(new Date()));
//        pool.shutdownNow();

        Thread t = new Thread(new MyCallableCreate());
        t.start();



//        File f = new File("D://file");
//        // 文件总数
//        final List<File> filePathsList = new ArrayList<File>();
//        File[] filePaths = f.listFiles();
//
//        for (File s : filePaths) {
//            if (s.getAbsolutePath().indexOf(".txt")>-1) {
//                filePathsList.add(s);
//                Map<String, File> fileMap = new HashMap<String, File>();
//                fileMap.put(s.getName(), s);
//                FileQueue.getQueue().add(fileMap);
//            }
//        }
//
//        // 创建一个文件观察器用于处理文件的格式
//        // 设置文件变化监听器
//        FileMonitor fileMonitor = new FileMonitor(1000);
//        fileMonitor.monitor("D://file", new FileListener());
//        fileMonitor.start();
//        CountDownLatch latch = new CountDownLatch(1024);
//        ExecutorService pool = Executors.newFixedThreadPool(4);
//
//        System.out.println("-------------文件读开始时间：" + sdf.format(new Date()));
////        for (int i = 0; i < filePathsList.size(); i++) {
////            pool.execute(new MyCallableCon(FileQueue.getQueue(), latch));
////        }
//
//        pool.execute(new MyCallableCon(FileQueue.getQueue(), latch));
//        try {
//            latch.await();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println("-------------文件读、写任务结束时间：" + sdf.format(new Date()));
//        pool.shutdownNow();

    }

    // 文件写线程
    // 模拟生产者进行文件创建,支持按条数、按文件大小创建文件
    public static class MyCallableCreate implements Runnable
    {
        //@Override
        public void run() {

            while (true) {
                long startTime = System.currentTimeMillis();
                String fileName = "d:" + File.separator + "reader" + File.separator + Thread.currentThread().getName() + System.currentTimeMillis();
                File temp = new File(fileName + ".temp");
                //File temp = new File(fileName + ".txt");

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                OutputStreamWriter osw = null;
                try {
                    osw = new OutputStreamWriter(new FileOutputStream(temp), "utf-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                }
                BufferedWriter bw = new BufferedWriter(osw);


                //String data = "Phonestring手机号|轨迹开始时间|轨迹结束时间|LACstring小区号|CELLstring基站号|LNGstring经度|LATstring纬度|O_LNGstring纠偏后的经度（用于google地图显示）|O_LATstring纠偏后的纬度（用于google地图显示）|PRECISIONstring基站信号覆盖范围（单位：米，半径）|ADDRESSstring地址";
                String data = "";
                int sum = 0;
                try {
                    //bw.write("+++++++++++++" + Thread.currentThread().getName() + " 线程开始写文件++++++++++++" + "\r");
                    while (true) {
                        String tel = InfoUtil.getTel();
                        //"武汉市","114.3-30.6"
                        String lng = InfoUtil.randomLonLat()[0];
                        String lat = InfoUtil.randomLonLat()[1];
                        data = tel + "|" + System.currentTimeMillis() +  "|" + (System.currentTimeMillis() + 1000) +"|LACstring小区号|CELLstring基站号|" + lng +"|" + lat +"|O_LNGstring纠偏后的经度（用于google地图显示）|O_LATstring纠偏后的纬度（用于google地图显示）|PRECISIONstring基站信号覆盖范围（单位：米，半径）|ADDRESSstring地址";
                        bw.write(data + "\r");
                        sum++;
                        //System.out.println(data);
                        if (temp.length() >= 2 * 1024 * 1024) {
                            break;
                        }
                    }

                    bw.flush();
                    osw.close();
                    bw.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                long endTime = System.currentTimeMillis();
                System.out.println(Thread.currentThread().getName() + " 线程写文件 ：" + temp.getName() + " 完毕" + " ,条数为 " + sum + " ,时间为 " + (endTime - startTime)/1000 + "s");
                try {
                    File resultFile = new File(fileName + ".txt");
                    FileUtils.copyFile(temp, resultFile);
                    temp.delete();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // 文件读线程
    // 支持对写入完成的文件，使用线程池进行消费。
    // 支持监听文件增量append并消费
    public static class MyCallableCon implements Runnable {
        private CountDownLatch latch;
        private String fileName = "";
        private BlockingQueue<Map<String, File>> queue;
        private File s = null;
        private FileInputStream fis = null;
        private BufferedReader br = null;
        private InputStreamReader isr = null;

        public MyCallableCon(BlockingQueue<Map<String, File>> queue, CountDownLatch latch)
        {
            this.queue = queue;
            this.latch = latch;
        }

        //@Override
        public void run(){
            while (true) {
                try {
                    long startTime = System.currentTimeMillis();
                    Map<String, File> map = queue.take();
                    int count = 0;
                    Set<String> set = map.keySet();
                    for (Iterator<String> iter = set.iterator(); iter.hasNext();) {
                        fileName = iter.next().toString();
                        s = map.get(fileName);
                        try {
                            fis = new FileInputStream(s);
                            isr = new InputStreamReader(fis, "utf-8");
                            br = new BufferedReader(isr);
                            String str = "";
                            while (br.readLine() != null) {
                                count++;
                            }
                            latch.countDown();
                        } catch (FileNotFoundException e) {
                            e.printStackTrace();
                        } catch (IOException e) {
                            e.printStackTrace();
                        } finally {
                            try {
                                fis.close();
                                isr.close();
                                br.close();
                                s.delete();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                    long endTime = System.currentTimeMillis();
                    System.out.println(Thread.currentThread().getName() + " 线程读文件 ：" + fileName + " 完毕" + " ,条数为 " + count + " ,时间为 " + (endTime - startTime)/1000 + "s");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}